package com.tyc.flink;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * 类描述
 *
 * @author tyc
 * @version 1.0
 * @date 2022-11-16 16:12:55
 */
public class StreamSourceFromKafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //从kafka读数据
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.1.1.145:30903,10.1.1.145:30904,10.1.1.145:30905");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"flink-me");
        FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<String>("test-flink", new SimpleStringSchema(), properties);
        // 指定偏移量
        myConsumer.setStartFromEarliest();
        DataStream<String> dataStreamSource = env.addSource(myConsumer);
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = dataStreamSource.flatMap(new MyStreamFlatMap()).keyBy(item -> item.f0).sum(1);
        // 打印数据
//        dataStreamSource.print();
        // 打印数据处理结果
        result.print();
        // 配置redis
        FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder()
                .setHost("10.1.1.91")
                .setPort(6379)
                .setPassword("111111")
                .setDatabase(1)
                .build();
        result.addSink(new RedisSink<>(config,new WordCountRedisSink()));
        env.execute();
    }

    private static class MyStreamFlatMap implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String record, Collector<Tuple2<String, Integer>> collector) throws Exception {
            collector.collect(new Tuple2<>(record,1));
        }
    }

    private static class WordCountRedisSink implements RedisMapper<Tuple2<String,Integer>> {

        // 存储的 key 名称与类型
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "word_count");
        }

        // 如何从数据中提取key值
        @Override
        public String getKeyFromData(Tuple2<String, Integer> stringIntegerTuple2) {
            return stringIntegerTuple2.f0;
        }

        // 如何从数据中提取key值
        @Override
        public String getValueFromData(Tuple2<String, Integer> stringIntegerTuple2) {
            return String.valueOf(stringIntegerTuple2.f1);
        }
    }

//    private static class MyStreamFlatMap implements FlatMapFunction<String, Tuple2<String, Integer>> {
//        @Override
//        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
//            String[] split = s.split("\\|");
//            if(split.length == 5 && "item_log".equals(split[3])){
//                String dataStr = split[4];
//                JSONObject jsonObject = JSONUtil.parseObj(dataStr);
//                collector.collect(new Tuple2<>(jsonObject.getStr("playerGuid"),jsonObject.getInt("itemCount")));
//            }
//        }
//    }
}
